package broker

import (
	"log"
	"net"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/yangyongzhen/mqtt-broker/internal/protocol/common"
	"github.com/yangyongzhen/mqtt-broker/internal/protocol/mqtt311"
	"github.com/yangyongzhen/mqtt-broker/internal/storage"
	"github.com/yangyongzhen/mqtt-broker/pkg/mqtt"
)

type Client struct {
	conn           net.Conn
	clientID       string
	info           *common.ClientInfo
	session        *storage.Session
	broker         *Broker
	packetReader   *mqtt.PacketReader
	writeChan      chan []byte
	closeChan      chan struct{}
	keepAliveTimer *time.Timer
	mu             sync.RWMutex
	connected      bool
	nextPacketID   uint16
	pendingAcks    map[uint16]*PendingMessage
}

type PendingMessage struct {
	Message   *common.Message
	Timestamp time.Time
	Retries   int
}

func NewClient(conn net.Conn, broker *Broker) *Client {
	return &Client{
		conn:         conn,
		broker:       broker,
		packetReader: mqtt.NewPacketReader(conn),
		writeChan:    make(chan []byte, 1000),
		closeChan:    make(chan struct{}),
		pendingAcks:  make(map[uint16]*PendingMessage),
		nextPacketID: 1,
	}
}

func (c *Client) Start() {
	go c.readLoop()
	go c.writeLoop()
	go c.retryLoop()
}

func (c *Client) readLoop() {
	defer c.Close()

	for {
		select {
		case <-c.closeChan:
			return
		default:
			packet, err := c.packetReader.ReadPacket()
			if err != nil {
				log.Printf("Read packet error: %v\n", err)
				return
			}

			c.handlePacket(packet)
		}
	}
}

func (c *Client) writeLoop() {
	defer c.Close()

	for {
		select {
		case data := <-c.writeChan:
			if _, err := c.conn.Write(data); err != nil {
				log.Printf("Write error: %v\n", err)
				return
			}
		case <-c.closeChan:
			return
		}
	}
}

func (c *Client) retryLoop() {
	ticker := time.NewTicker(5 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			c.retryPendingMessages()
		case <-c.closeChan:
			return
		}
	}
}

func (c *Client) handlePacket(packet common.Packet) {
	switch p := packet.(type) {
	case *mqtt311.ConnectPacket:
		c.handleConnect(p)
	case *mqtt311.PublishPacket:
		c.handlePublish(p)
	case *mqtt311.SubscribePacket:
		c.handleSubscribe(p)
	case *mqtt.PingReqPacket:
		c.handlePingReq()
	case *mqtt.DisconnectPacket:
		c.handleDisconnect()
	}
}

func (c *Client) handleConnect(packet *mqtt311.ConnectPacket) {
	// 验证协议版本
	if packet.ProtocolVersion != 4 {
		c.sendConnack(false, 0x01) // 不支持的协议版本
		c.Close()
		return
	}

	// 生成客户端ID（如果为空）
	clientID := packet.ClientID
	if clientID == "" {
		if !packet.CleanSession {
			c.sendConnack(false, 0x02) // 标识符被拒绝
			c.Close()
			return
		}
		clientID = uuid.New().String()
	}

	// 设置客户端信息
	c.clientID = clientID
	c.info = &common.ClientInfo{
		ClientID:        clientID,
		Username:        packet.Username,
		Password:        string(packet.Password),
		KeepAlive:       packet.KeepAlive,
		CleanSession:    packet.CleanSession,
		WillTopic:       packet.WillTopic,
		WillMessage:     packet.WillMessage,
		WillQoS:         packet.WillQoS,
		WillRetain:      packet.WillRetain,
		ProtocolVersion: common.MQTT311,
	}

	// 认证
	if !c.broker.authenticateClient(c.info) {
		c.sendConnack(false, 0x04) // 用户名或密码错误
		c.Close()
		return
	}

	// 会话处理
	sessionPresent := false
	if !packet.CleanSession {
		// 尝试恢复会话
		session, err := c.broker.store.LoadSession(clientID)
		if err == nil && session != nil {
			c.session = session
			sessionPresent = true
		}
	}

	if c.session == nil {
		c.session = &storage.Session{
			ClientID:      clientID,
			CleanSession:  packet.CleanSession,
			Subscriptions: make(map[string]*common.Subscription),
			PendingAcks:   make(map[uint16]*common.Message),
			LastSeen:      time.Now(),
		}
	}

	// 添加到broker
	c.broker.AddClient(clientID, c)
	c.connected = true

	// 发送CONNACK
	c.sendConnack(sessionPresent, 0x00)

	// 设置保活定时器
	if packet.KeepAlive > 0 {
		c.resetKeepAliveTimer(time.Duration(packet.KeepAlive) * time.Second * 3 / 2)
	}

	// 发送保留消息
	c.sendRetainedMessages()

	// 重发未确认的消息
	c.resendPendingMessages()
}

func (c *Client) handlePublish(packet *mqtt311.PublishPacket) {
	c.resetKeepAliveTimer(time.Duration(c.info.KeepAlive) * time.Second * 3 / 2)

	message := &common.Message{
		Topic:     packet.TopicName,
		Payload:   packet.Payload,
		QoS:       packet.QoS,
		Retain:    packet.Retain,
		PacketID:  packet.PacketID,
		Timestamp: time.Now(),
	}

	switch packet.QoS {
	case common.QoS0:
		// QoS 0: 最多一次传递
		c.broker.PublishMessage(message)

	case common.QoS1:
		// QoS 1: 至少一次传递
		c.broker.PublishMessage(message)
		c.sendPuback(packet.PacketID)

	case common.QoS2:
		// QoS 2: 恰好一次传递
		// 第一步：发送PUBREC
		c.sendPubrec(packet.PacketID)
		// 存储消息等待PUBREL
		c.session.PendingAcks[packet.PacketID] = message
	}
}

func (c *Client) handleSubscribe(packet *mqtt311.SubscribePacket) {
	c.resetKeepAliveTimer(time.Duration(c.info.KeepAlive) * time.Second * 3 / 2)

	returnCodes := make([]byte, len(packet.Subscriptions))

	for i, subscription := range packet.Subscriptions {
		// 验证主题过滤器
		if !c.isValidTopicFilter(subscription.Topic) {
			returnCodes[i] = 0x80 // 失败
			continue
		}

		// 添加订阅
		c.session.Subscriptions[subscription.Topic] = &subscription
		c.broker.topicManager.Subscribe(c.clientID, &subscription)

		// 保存会话
		c.broker.store.SaveSession(c.clientID, c.session)

		returnCodes[i] = byte(subscription.QoS)

		// 发送匹配的保留消息
		c.sendMatchingRetainedMessages(subscription.Topic)
	}

	// 发送SUBACK
	c.sendSuback(packet.PacketID, returnCodes)
}

func (c *Client) handlePingReq() {
	c.resetKeepAliveTimer(time.Duration(c.info.KeepAlive) * time.Second * 3 / 2)
	c.sendPingResp()
}

func (c *Client) handleDisconnect() {
	c.Close()
}

func (c *Client) sendConnack(sessionPresent bool, returnCode byte) {
	packet := &mqtt311.ConnackPacket{
		SessionPresent: sessionPresent,
		ReturnCode:     returnCode,
	}
	packet.MessageType = common.CONNACK

	data, err := packet.Encode()
	if err != nil {
		log.Printf("Encode CONNACK error: %v\n", err)
		return
	}

	c.Send(data)
}

func (c *Client) sendPuback(packetID uint16) {
	data := []byte{0x40, 0x02, byte(packetID >> 8), byte(packetID & 0xFF)}
	c.Send(data)
}

func (c *Client) sendPubrec(packetID uint16) {
	data := []byte{0x50, 0x02, byte(packetID >> 8), byte(packetID & 0xFF)}
	c.Send(data)
}

func (c *Client) sendSuback(packetID uint16, returnCodes []byte) {
	packet := &mqtt311.SubackPacket{
		PacketID:    packetID,
		ReturnCodes: returnCodes,
	}
	packet.MessageType = common.SUBACK

	data, err := packet.Encode()
	if err != nil {
		log.Printf("Encode SUBACK error: %v\n", err)
		return
	}

	c.Send(data)
}

func (c *Client) sendPingResp() {
	data := []byte{0xD0, 0x00}
	c.Send(data)
}

func (c *Client) PublishToClient(message *common.Message, qos common.QoS) {
	if !c.connected {
		return
	}

	packet := &mqtt311.PublishPacket{
		TopicName: message.Topic,
		Payload:   message.Payload,
		PacketID:  0,
	}
	packet.MessageType = common.PUBLISH
	packet.QoS = qos
	packet.Retain = message.Retain

	if qos > common.QoS0 {
		packet.PacketID = c.nextPacketID
		c.nextPacketID++
		if c.nextPacketID == 0 {
			c.nextPacketID = 1
		}

		// 存储待确认消息
		c.pendingAcks[packet.PacketID] = &PendingMessage{
			Message:   message,
			Timestamp: time.Now(),
			Retries:   0,
		}
	}

	data, err := packet.Encode()
	if err != nil {
		log.Printf("Encode PUBLISH error: %v\n", err)
		return
	}

	c.Send(data)
}

func (c *Client) Send(data []byte) {
	select {
	case c.writeChan <- data:
	case <-c.closeChan:
	}
}

func (c *Client) Close() {
	c.mu.Lock()
	defer c.mu.Unlock()

	if !c.connected {
		return
	}

	c.connected = false

	// 发送遗嘱消息
	if c.info != nil && c.info.WillTopic != "" {
		willMessage := &common.Message{
			Topic:     c.info.WillTopic,
			Payload:   c.info.WillMessage,
			QoS:       c.info.WillQoS,
			Retain:    c.info.WillRetain,
			Timestamp: time.Now(),
		}
		c.broker.PublishMessage(willMessage)
	}

	// 保存会话
	if c.session != nil && !c.session.CleanSession {
		c.session.LastSeen = time.Now()
		c.broker.store.SaveSession(c.clientID, c.session)
	}

	// 停止定时器
	if c.keepAliveTimer != nil {
		c.keepAliveTimer.Stop()
	}

	close(c.closeChan)
	c.conn.Close()

	if c.broker != nil {
		c.broker.RemoveClient(c.clientID)
	}
}

func (c *Client) resetKeepAliveTimer(duration time.Duration) {
	if c.keepAliveTimer != nil {
		c.keepAliveTimer.Stop()
	}

	if duration > 0 {
		c.keepAliveTimer = time.AfterFunc(duration, func() {
			log.Printf("Client %s keep alive timeout\n", c.clientID)
			c.Close()
		})
	}
}

func (c *Client) isValidTopicFilter(topic string) bool {
	// 实现主题过滤器验证逻辑
	if len(topic) == 0 {
		return false
	}

	// 检查通配符使用是否正确
	for i, char := range topic {
		if char == '#' {
			// # 必须是最后一个字符，且前面必须是 / 或者是第一个字符
			if i != len(topic)-1 {
				return false
			}
			if i > 0 && topic[i-1] != '/' {
				return false
			}
		} else if char == '+' {
			// + 前后必须是 / 或者是开头/结尾
			if i > 0 && topic[i-1] != '/' {
				return false
			}
			if i < len(topic)-1 && topic[i+1] != '/' {
				return false
			}
		}
	}

	return true
}

func (c *Client) sendRetainedMessages() {
	// 为所有订阅发送保留消息
	for topic := range c.session.Subscriptions {
		c.sendMatchingRetainedMessages(topic)
	}
}

func (c *Client) sendMatchingRetainedMessages(topicFilter string) {
	// 获取匹配的保留消息
	retainedMessages := c.broker.getMatchingRetainedMessages(topicFilter)
	for _, message := range retainedMessages {
		subscription := c.session.Subscriptions[topicFilter]
		qos := message.QoS
		if subscription.QoS < qos {
			qos = subscription.QoS
		}
		c.PublishToClient(message, qos)
	}
}

func (c *Client) resendPendingMessages() {
	// 重发未确认的消息
	for _, pendingMsg := range c.session.PendingAcks {
		message := pendingMsg
		c.PublishToClient(message, message.QoS)
	}
}

func (c *Client) retryPendingMessages() {
	c.mu.Lock()
	defer c.mu.Unlock()

	now := time.Now()
	for packetID, pending := range c.pendingAcks {
		if now.Sub(pending.Timestamp) > 30*time.Second {
			if pending.Retries < 3 {
				// 重试
				pending.Retries++
				pending.Timestamp = now
				c.PublishToClient(pending.Message, pending.Message.QoS)
			} else {
				// 超过重试次数，删除
				delete(c.pendingAcks, packetID)
			}
		}
	}
}
